package com.kdpujie;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


/**
 * 每间隔1分钟，统计过去5分钟的UV，PV
 * 
 */
public class PvUvJob {
    public static void main(String[] args) {
            // 初始化执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<UserEvent> eventSource = env.addSource(new EventSource());
        DataStream<Tuple4<Long, Long, Integer, Integer>> uvAndPv = eventSource.windowAll(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(1)))
        .allowedLateness(Time.minutes(5))
        .process(new ProcessAllWindowFunction<UserEvent,Tuple4<Long, Long, Integer, Integer>, TimeWindow>(){

            @Override
            public void process(Context context, Iterable<UserEvent> elements, Collector<Tuple4<Long, Long, Integer, Integer>> out) throws Exception {
                Integer pv = 0;
                TimeWindow window = context.window();
                Set<Integer> userIds = new HashSet<>();
                Iterator<UserEvent> iterator = elements.iterator();
                while(iterator.hasNext()) {
                    UserEvent e = iterator.next();
                    pv++;
                    userIds.add(e.getUserId());
                }
                out.collect(new Tuple4<>(window.getStart(), window.getEnd(), pv, userIds.size()));
            }
            
        });
        uvAndPv.print();

        try {
            env.execute("computing pv-uv");
        } catch (Exception e) {
            e.printStackTrace();
        }
       
    }
}
